// ---------------------------------------------------------------------
// exemple-threads-mq.c
// ---------------------------------------------------------------------

#include <pthread.h>
#include <sched.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <sys/time.h>
#include <mqueue.h>
#include <sys/stat.h>
#include <errno.h>
#include <string.h>
#include <time.h>

/* Threads */
static void *MainThread (void *);
static void *AnotherThread (void *);
static void *TroisThread (void *);
static void *QuatreThread (void *);

/* Defines */
#define MAIN_QNAME "/MAINQUEUE"
pthread_mutex_t wait_mutex;

typedef struct sMSG_versErreur{
	int source;
	int message;
	time_t heure;
	int numLot;
	int cartonEnCours;
}MSG_versErreur;

int mq_test(void)
{
 pthread_t mainThread, anotherThread, troisThread, quatreThread;
  
 printf ("Creating threads .. \n");
 pthread_create (&mainThread,    NULL, &MainThread,    NULL);
sleep(3);
 pthread_create (&troisThread, NULL, &TroisThread, NULL);
sleep(3);

 pthread_create (&quatreThread, NULL, &QuatreThread, NULL);

sleep(20);
 pthread_create (&anotherThread, NULL, &AnotherThread, NULL);

 pthread_mutex_init (&wait_mutex, NULL);
 
 pthread_join (mainThread,    NULL);
 pthread_join (anotherThread, NULL);
pthread_join (troisThread, NULL);
 
 return 1;
}

#define BUFFER_SIZE 10000

/* Main thread .. Waiting for messages */
static void *MainThread (void *args)
{
 mqd_t queue_handle;
 char buffer[BUFFER_SIZE];
 int bytes_read;
 struct mq_attr attr;
 unsigned int sender;
  attr.mq_maxmsg = 20;
  attr.mq_msgsize = 128;
  attr.mq_flags   = 0;
 
 printf ("[MainThread] Inside main thread \n");

 // Let the other thread wait till I am ready! 
 pthread_mutex_lock (&wait_mutex);
 
 // Clear the buffer
 memset (buffer, 0, BUFFER_SIZE);

 // Detach the thread
 pthread_detach (pthread_self());
 
 // unlink the queue if it exisits - debug
 mq_unlink (MAIN_QNAME);

 printf ("[MainThread]Opening MQ \n"); 
struct mq_attr lesAttr={0,1,sizeof(MSG_versErreur),0};
 queue_handle=  mq_open(MAIN_QNAME, O_RDWR | O_CREAT, 0777 , &lesAttr);

 if (queue_handle == -1)
 {
  perror ("[MainThread] Error Opening MQ: ");
  return 0;
 } 
 
 printf ("[MainThread] Waiting for messages ... \n"); 
 pthread_mutex_unlock (&wait_mutex);
 
 for (; ;)
 {
MSG_versErreur messageRecu;
  bytes_read = mq_receive(queue_handle, (char *) &messageRecu, sizeof(MSG_versErreur), &sender);

  if (bytes_read == -1)
  {
   perror("[MainThread] Failed to recieve:");
   return 0;
  }
  else
  {
   //MSG_versErreur * messageRecu = (MSG_versErreur *) buffer ;
   //printf("message %s \n", buffer);
   printf ("[MainThread] source: %d , message : %d, numlot %d, carton %d \n", messageRecu.source, messageRecu.message, messageRecu.numLot, messageRecu.cartonEnCours);   
   //printf("Taille : %d et tot %d /n",sizeof(*messageRecu), sizeof( MSG_versErreur));
   // Get the MQ attributes
   struct mq_attr msgq_attr2;
   mq_getattr(queue_handle, &msgq_attr2);
   printf("[MainThread] Queue \"%s\":\n"
      "\t- stores at most %ld messages\n"
      "\t- large at most %ld bytes each\n"
      "\t- currently holds %ld messages\n",
       MAIN_QNAME, 
       msgq_attr2.mq_maxmsg, 
       msgq_attr2.mq_msgsize, 
       msgq_attr2.mq_curmsgs);   

   // Clear buffer and sleep to block for some time t and see 
   // if you get all the messages!
   memset(buffer, 0, BUFFER_SIZE); 
   sleep(5);
  }  
 } 

 mq_close (queue_handle);
}

static void *QuatreThread (void *args)
{
 mqd_t queue_handle;
 char buffer[BUFFER_SIZE];
 int bytes_read;
 struct mq_attr attr;
 unsigned int sender;
 unsigned int msgprio = 1;
 int count;
 
 printf ("[QuatreThread] Inside QuatreThread thread \n");
 queue_handle=  mq_open(MAIN_QNAME, O_RDWR);
 if (queue_handle == -1)
 {
  perror ("[QuatreThread] Error Opening MQ:");
  return 0;
 }
 for (; ;)
 {
MSG_versErreur messageRecu;
  bytes_read = mq_receive(queue_handle, (char *) &messageRecu, sizeof(MSG_versErreur), &sender);

  if (bytes_read == -1)
  {
   perror("[QuatreThread] Failed to recieve:");
   return 0;
  }
  else
  {
   //MSG_versErreur * messageRecu = (MSG_versErreur *) buffer ;
   //printf("message %s \n", buffer);
   printf ("QuatreThread] source: %d , message : %d, numlot %d, carton %d \n", messageRecu.source, messageRecu.message, messageRecu.numLot, messageRecu.cartonEnCours);   
   //printf("Taille : %d et tot %d /n",sizeof(*messageRecu), sizeof( MSG_versErreur));
   // Get the MQ attributes
   struct mq_attr msgq_attr2;
   mq_getattr(queue_handle, &msgq_attr2);
   printf("[QuatreThread] Queue \"%s\":\n"
      "\t- stores at most %ld messages\n"
      "\t- large at most %ld bytes each\n"
      "\t- currently holds %ld messages\n",
       MAIN_QNAME, 
       msgq_attr2.mq_maxmsg, 
       msgq_attr2.mq_msgsize, 
       msgq_attr2.mq_curmsgs);   

   // Clear buffer and sleep to block for some time t and see 
   // if you get all the messages!
   memset(buffer, 0, BUFFER_SIZE); 
   sleep(5);
  }  
 mq_close (queue_handle); 
 return 0;
}
}


static void *TroisThread (void *args)
{
 mqd_t queue_handle;
 char buffer[BUFFER_SIZE];
 int bytes_read;
 struct mq_attr attr;
 unsigned int sender;
 unsigned int msgprio = 1;
 int count;
 
 printf ("[TroisThread] Inside Trois thread \n");
 queue_handle=  mq_open(MAIN_QNAME, O_RDWR);
 if (queue_handle == -1)
 {
  perror ("[TroisThread] Error Opening MQ:");
  return 0;
 }
 for (; ;)
 {
MSG_versErreur messageRecu;
  bytes_read = mq_receive(queue_handle, (char *) &messageRecu, sizeof(MSG_versErreur), &sender);

  if (bytes_read == -1)
  {
   perror("[TroisThread] Failed to recieve:");
   return 0;
  }
  else
  {
   //MSG_versErreur * messageRecu = (MSG_versErreur *) buffer ;
   //printf("message %s \n", buffer);
   printf ("TroisThread] source: %d , message : %d, numlot %d, carton %d \n", messageRecu.source, messageRecu.message, messageRecu.numLot, messageRecu.cartonEnCours);   
   //printf("Taille : %d et tot %d /n",sizeof(*messageRecu), sizeof( MSG_versErreur));
   // Get the MQ attributes
   struct mq_attr msgq_attr2;
   mq_getattr(queue_handle, &msgq_attr2);
   printf("[TroisThread] Queue \"%s\":\n"
      "\t- stores at most %ld messages\n"
      "\t- large at most %ld bytes each\n"
      "\t- currently holds %ld messages\n",
       MAIN_QNAME, 
       msgq_attr2.mq_maxmsg, 
       msgq_attr2.mq_msgsize, 
       msgq_attr2.mq_curmsgs);   

   // Clear buffer and sleep to block for some time t and see 
   // if you get all the messages!
   memset(buffer, 0, BUFFER_SIZE); 
   sleep(5);
  }  
 mq_close (queue_handle); 
 return 0;
}
}

#define MAX_SEND_BUFFER 10000 

static void *AnotherThread (void *args)
{
 mqd_t queue_handle;
 char * buffer;
 unsigned int msgprio = 1;
 int count;
 
 printf ("[AnotherThread] Inside Another thread \n");
 pthread_mutex_lock (&wait_mutex);
 queue_handle=  mq_open(MAIN_QNAME, O_RDWR);
 if (queue_handle == -1)
 {
  perror ("[AnotherThread] Error Opening MQ:");
  return 0;
 }
 for (count = 0; count < 10; count++)
 {
  MSG_versErreur messageEnvoie;
  messageEnvoie.source = 2*count ;
  messageEnvoie.message = 3*count ;
  messageEnvoie.heure = time(NULL);
  messageEnvoie.numLot = 4*count ;
  messageEnvoie.cartonEnCours = 5*count ;
 // printf ("[AnotherThread] source: %d , message : %d, numlot %d, carton %d \n\n", messageEnvoie.source, messageEnvoie.message, messageEnvoie.numLot, messageEnvoie.cartonEnCours);   
 // printf("tot %d, 1 %d, 2 %d,3 %d,4 %d,5 %d \n\n", sizeof(MSG_versErreur), sizeof(int), sizeof(int), sizeof(time_t), sizeof(int), sizeof(int));
  //buffer = (char*) &messageEnvoie;
 // printf("message anotherThread %s \n", buffer);
  if (0 != mq_send (queue_handle, (const char *) &messageEnvoie, sizeof(MSG_versErreur), msgprio))
  {
   perror ("[AnotherThread] Sending:");
   mq_close (queue_handle); 
   pthread_mutex_unlock (&wait_mutex);
   return 0;
  }
 } 
 pthread_mutex_unlock (&wait_mutex); 
 mq_close (queue_handle); 
 return 0;
}

int main ()
{
 return mq_test();
}

 
